/*
 * ThreadedQueu.h
 *
 *  Created on: Oct 6, 2011
 *      Author: fremm
 */

#ifndef THREADEDQUEU_H_
#define THREADEDQUEU_H_


#include <stdio.h>
#include <stdlib.h>
#include "ThreadedQueue.h"
#include "pthread.h"
#include "string.h"


typedef struct queueCDT {
   queueElementT contents[MAX_QUEUE_SIZE];
   int front;
   int count;
   bool termination;
   pthread_mutex_t 	mutex;
   pthread_cond_t	conditionFull;
   pthread_cond_t 	conditionEmpty;
} queueCDT;

queueADT QueueCreate(void)
{
   queueADT queue;

   queue = (queueADT)malloc(sizeof(queueCDT));

   if (queue == NULL) {
      fprintf(stderr, "Insufficient Memory for new Queue.\n");
      exit(ERROR_MEMORY);  /* Exit program, returning error code. */
   }

   queue->front = 0;
   queue->count = 0;
   queue->termination = false;

   pthread_mutex_init(&queue->mutex, NULL);
   pthread_cond_init(&queue->conditionFull, NULL);
   pthread_cond_init(&queue->conditionEmpty, NULL);

   return queue;
}

void QueueDestroy(queueADT queue)
{
   free(queue);
}

void push(queueADT queue, queueElementT element)
{
   int newElementIndex;

   pthread_mutex_lock(&queue->mutex);

   while ( queue->count >= MAX_QUEUE_SIZE)
   {
      pthread_cond_wait(&queue->conditionFull, &queue->mutex);
   }

   bool is_empty = queue->count == 0;

   newElementIndex = (queue->front + queue->count) % MAX_QUEUE_SIZE;
   queue->contents[newElementIndex] = (char*)malloc(sizeof(char)*strlen(element)+1);
   memcpy(queue->contents[newElementIndex], element, sizeof(char)*strlen(element)+1);

   queue->count++;
   if ( is_empty)
      pthread_cond_signal(&queue->conditionEmpty);

   pthread_mutex_unlock(&queue->mutex);

}

void wait_and_pop(queueADT queue, queueElementT* elem)
{
   pthread_mutex_lock(&queue->mutex);


   while(queue->count == 0 )
   {
      pthread_cond_wait(&queue->conditionEmpty, &queue->mutex);
   }

   bool queueFull = (queue->count == MAX_QUEUE_SIZE) ? true : false;

   *elem = queue->contents[queue->front];

   if ( queue->count == 1)
   {
      if ( memcmp(*elem, "#", 2) == 0)
      {
         queue->termination = true;
         pthread_cond_broadcast(&queue->conditionEmpty);
      }
   }
   else
   {
      queue->front++;
      queue->front %= MAX_QUEUE_SIZE;

      queue->count--;
   }

   if ( queueFull)
   {
      pthread_cond_signal(&queue->conditionFull);
   }


   pthread_mutex_unlock(&queue->mutex);

}

bool empty(queueADT queue)
{
   pthread_mutex_lock(&queue->mutex);
   bool ret = queue->count == 0;
   pthread_mutex_unlock(&queue->mutex);
   return ret;
}

bool isTermination(queueADT queue)
{
   pthread_mutex_lock(&queue->mutex);
   bool term = queue->termination;
   pthread_mutex_unlock(&queue->mutex);
   return term;
}

bool full(queueADT queue)
{
   pthread_mutex_lock(&queue->mutex);
   bool ret = queue->count == MAX_QUEUE_SIZE;
   pthread_mutex_unlock(&queue->mutex);
   return ret;
}

#endif /* THREADEDQUEU_H_ */
